import com.zuikaku.pojo.People;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketSourceDemo {

    public static void main(String[] args) {

        //1.创建流执行环境(指定为带webui的环境)，默认的webui端口为8081
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.指定数据流来源source，此时source来自于目标ip:8888的socket连接，可用netcat来创建该socket连接
        DataStream<String> source = env.socketTextStream("139.224.231.81",8888);
        source.print("源数据(source)");
        DataStream<People> sink = source.flatMap(new FlatMapFunction<String, People>() {
            @Override
            public void flatMap(String s, Collector<People> collector) throws Exception {
                String[] temp = s.split(",");
                String name = temp[0];
                int age = Integer.parseInt(temp[1]);
                float weight =Float.parseFloat(temp[2]);
                People p = new People(name,age,weight);
                collector.collect(p);
            }
        });
        sink.print("处理后(sink)");
        //DataStream需要调用execute,可以取个名称
        try {
            env.execute("socket source stream job");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
